//
// ThreadPool.cpp
//
// $Id: //poco/1.4/Foundation/src/ThreadPool.cpp#2 $
//
// Library: Foundation
// Package: Threading
// Module:  ThreadPool
//
// Copyright (c) 2004-2006, Applied Informatics Software Engineering GmbH.
// and Contributors.
//
// SPDX-License-Identifier:	BSL-1.0
//


#include "Poco/ThreadPool.h"
#include "Poco/Runnable.h"
#include "Poco/Thread.h"
#include "Poco/Event.h"
#include "Poco/ErrorHandler.h"
#include "Poco/Exception.h"
#include <sstream>
#include <ctime>
#include <time.h>

namespace Poco {

/// 伪线程，if no error occured or manually stoped, it will never stop
class PooledThread: public Runnable
{
public:
	PooledThread(const std::string& name, int stackSize = POCO_THREAD_STACK_SIZE);
	~PooledThread();

	void start(ThreadPriority priority, Runnable& target);
		/// when target is not null, throw ThreadAlreadyRunException
	void start(ThreadPriority priority, Runnable& target, const std::string& name);
		/// when target is not null, throw ThreadAlreadyRunException
	bool idle();
	int idleTime();
	void join();
	//void activate();
	void release();
		/// finish run() immediately.
	void run();
private:
	friend class ThreadPool;
private:

	volatile std::time_t m_idleTime;
	volatile std::time_t m_workingTime;
	Runnable*            m_pTarget;
	std::string          m_name;
	Thread               m_thread;
	Event                m_targetReady;			/// auto reset
	Event                m_targetCompleted;
	//FastMutex            m_mutex;				/// not needed by threadpool
	bool				 m_idle;
	bool 				 m_isReenterable;		/// 用于一个target运行结束后，再次运行另一个target，而不用重新新建线程
};


PooledThread::PooledThread(const std::string& name, int stackSize): 
	m_idleTime(0),
	m_workingTime(0),
	m_pTarget(0), 
	m_name(name), 
	m_thread(name),
	m_targetReady(true),
	m_targetCompleted(false),
	m_idle(true),
	m_isReenterable(false)
{
	poco_assert_dbg (stackSize >= 0);
	m_thread.setStackSize(stackSize);
	m_idleTime = std::time(NULL);
	m_thread.start(*this);
}


PooledThread::~PooledThread()
{
}

void PooledThread::start(ThreadPriority priority, Runnable& target)
{
	//FastMutex::ScopedLock lock(m_mutex);
	if (m_isReenterable)
	{
		m_targetCompleted.reset();
		m_isReenterable = false;
	}
	if(!m_idle)
	{
		throw ThreadAlreadyRunException();
	}
	m_pTarget = &target;
	m_thread.setPriority(priority);
	m_targetReady.set();
}


void PooledThread::start(ThreadPriority priority, Runnable& target, const std::string& name)
{
	//FastMutex::ScopedLock lock(m_mutex);
	if (m_isReenterable)
	{
		m_targetCompleted.reset();
		m_isReenterable = false;
	}
	std::string fullName(name);
	if (name.empty())
	{
		fullName = m_name;
	}
	else
	{
		fullName.append(" (");
		fullName.append(m_name);
		fullName.append(")");
	}
	m_thread.setName(fullName);
	m_thread.setPriority(priority);
	
	if(!m_idle)
	{
		throw ThreadAlreadyRunException();
	}

	m_pTarget = &target;
	m_targetReady.set();
}


inline bool PooledThread::idle()
{
	//FastMutex::ScopedLock lock(m_mutex);
	return m_idle;
}


int PooledThread::idleTime()
{
	//FastMutex::ScopedLock lock(m_mutex);
	return (int) (time(NULL) - m_idleTime);
}


void PooledThread::join()
{
	//m_mutex.lock();
	Runnable* pTarget = m_pTarget;
	//m_mutex.unlock();
	if (pTarget)
		m_targetCompleted.wait();
	m_thread.join();
}


void PooledThread::release()
{
// 	const long JOIN_TIMEOUT = 1000;
// 	
// 	m_mutex.lock();
// 	m_pTarget = 0;
// 	m_mutex.unlock();
// 	// In case of a statically allocated thread pool (such
// 	// as the default thread pool), Windows may have already
// 	// terminated the thread before we got here.
// 	if (m_thread.isRunning())
// 		m_targetReady.set();
// 
// 	if (m_thread.tryJoin(JOIN_TIMEOUT))
// 	{
// 		// 出错 销毁
// 		delete this;
// 	}
	delete this;
}


void PooledThread::run()
{
	for (;;)
	{
		m_targetReady.wait();	/// auto resetted
		//FastMutex::ScopedLock lock(m_mutex);
		m_idle = false;
		if (m_pTarget)
		{
			m_workingTime = time(NULL);
			try
			{
				m_pTarget->run();
			}
			catch (Exception& exc)
			{
				ErrorHandler::handle(exc);
			}
			catch (std::exception& exc)
			{
				ErrorHandler::handle(exc);
			}
			catch (...)
			{
				ErrorHandler::handle();
			}
			//break;
		}
		m_pTarget  = 0;
		m_idleTime = time(NULL);
		m_targetCompleted.set();
		m_thread.setName(m_name);
		m_thread.setPriority(PRIO_NORMAL);
		m_idle = true;
		m_isReenterable = true;
	}
}


ThreadPool::ThreadPool(int minCapacity,
	int maxCapacity,
	int idleTime,
	int stackSize): 
	m_minCapacity(minCapacity), 
	m_maxCapacity(maxCapacity), 
	m_idleTime(idleTime),
	m_serial(0),
	m_age(0),
	m_stackSize(stackSize)
{
	poco_assert (minCapacity >= 1 && maxCapacity >= minCapacity && idleTime > 0);

	for (int i = 0; i < m_minCapacity; i++)
	{
		PooledThread* pThread = createThread();
		m_idleThreads.push_back(pThread);
	}
}


ThreadPool::ThreadPool(const std::string& name,
	int minCapacity,
	int maxCapacity,
	int idleTime,
	int stackSize):
	m_name(name),
	m_minCapacity(minCapacity), 
	m_maxCapacity(maxCapacity), 
	m_idleTime(idleTime),
	m_serial(0),
	m_age(0),
	m_stackSize(stackSize)
{
	poco_assert (minCapacity >= 1 && maxCapacity >= minCapacity && idleTime > 0);

	for (int i = 0; i < m_minCapacity; i++)
	{
		PooledThread* pThread = createThread();
		m_idleThreads.push_back(pThread);
	}
}


ThreadPool::~ThreadPool()
{
	try
	{
		stopAll();
		clearIdle();
	}
	catch (...)
	{
		poco_unexpected();
	}
}


void ThreadPool::addCapacity(int n)
{
	FastMutex::ScopedLock lock(m_mutex);

	poco_assert (m_maxCapacity + n >= m_minCapacity);
	m_maxCapacity += n;
}


int ThreadPool::capacity() const
{
	FastMutex::ScopedLock lock(m_mutex);
	return m_maxCapacity;
}


int ThreadPool::available() const
{
	FastMutex::ScopedLock lock(m_mutex);

	return m_idleThreads.size();
}


int ThreadPool::working() const
{
	FastMutex::ScopedLock lock(m_mutex);
	return m_workingThreads.size();
}


int ThreadPool::allocated() const
{
	FastMutex::ScopedLock lock(m_mutex);

	return m_idleThreads.size() + m_workingThreads.size();
}


void ThreadPool::start(Runnable& target)
{
	getThread()->start(PRIO_NORMAL, target);
}


void ThreadPool::start(Runnable& target, const std::string& name)
{
	getThread()->start(PRIO_NORMAL, target, name);
}


void ThreadPool::startWithPriority(ThreadPriority priority, Runnable& target)
{
	getThread()->start(priority, target);
}


void ThreadPool::startWithPriority(ThreadPriority priority, Runnable& target, const std::string& name)
{
	getThread()->start(priority, target, name);
}


void ThreadPool::stopAll()
{
	FastMutex::ScopedLock lock(m_mutex);
	for (ThreadList::iterator it = m_workingThreads.begin(); it != m_workingThreads.end(); ++it)
	{
		(*it)->release();
	}
	m_workingThreads.clear();
}

void ThreadPool::clearIdle()
{
	FastMutex::ScopedLock lock(m_mutex);
	for (ThreadList::iterator it = m_idleThreads.begin(); it != m_idleThreads.end(); ++it)
	{
		(*it)->release();
	}
	m_idleThreads.clear();
}


void ThreadPool::joinAll()
{
	FastMutex::ScopedLock lock(m_mutex);

	for (ThreadList::iterator it = m_workingThreads.begin(); it != m_workingThreads.end(); ++it)
	{
		(*it)->join();
		m_idleThreads.push_back(*it);
	}
}

void ThreadPool::joinTimeoutWorkingThread(int second)
{
	FastMutex::ScopedLock lock(m_mutex);
	for(auto it = m_workingThreads.begin(); it != m_workingThreads.end();)
	{
		if ((*it)->m_workingTime != 0 && time(NULL) - (*it)->m_workingTime >= second)
		{
			(*it)->release();
			m_workingThreads.erase(it++);
			continue;
		}
		++it;
	}
	/// 整理已运行完毕的线程
	for(auto it = m_workingThreads.begin(); it != m_workingThreads.end();)
	{
		if ((*it)->idle() && (*it)->m_pTarget == nullptr)
		{
			m_idleThreads.push_back(*it);
			m_workingThreads.erase(it++);
			continue;
		}
		it++;
	}
}

// void ThreadPool::housekeep()
// {
// 	FastMutex::ScopedLock lock(m_mutex);
// 	/// 整理已运行完毕的线程
// 	for(auto it = m_workingThreads.begin(); it != m_workingThreads.end();)
// 	{
// 		if ((*it)->idle() && (*it)->m_pTarget == nullptr)
// 		{
// 			m_idleThreads.push_back(*it);
// 			m_workingThreads.erase(it++);
// 			continue;
// 		}
// 		it++;
// 	}
// }

PooledThread* ThreadPool::getThread()
{
	FastMutex::ScopedLock lock(m_mutex);
	/// 整理已运行完毕的线程
	for(auto it = m_workingThreads.begin(); it != m_workingThreads.end();)
	{
		if ((*it)->idle() && (*it)->m_pTarget == nullptr)
		{
			m_idleThreads.push_back(*it);
			m_workingThreads.erase(it++);
			continue;
		}
		it++;
	}
	if (m_idleThreads.size() > 0)
	{
		PooledThread* pThread = m_idleThreads.front();
		m_idleThreads.pop_front();
		m_workingThreads.push_back(pThread);
		return pThread;
	}
	
	if (m_workingThreads.size() + m_idleThreads.size() < m_maxCapacity)
	{
		PooledThread* pThread = createThread();
		m_workingThreads.push_back(pThread);
		return pThread;
	}
	else
	{
		throw NoThreadAvailableException();
	}
}


PooledThread* ThreadPool::createThread()
{
	std::ostringstream name;
	name << m_name << "[#" << ++m_serial << "]";
	m_serial = m_serial%2000000;
	return new PooledThread(name.str(), m_stackSize);
}


class ThreadPoolSingletonHolder
{
public:
	ThreadPoolSingletonHolder()
	{
		m_pPool = 0;
	}
	~ThreadPoolSingletonHolder()
	{
		delete m_pPool;
	}
	ThreadPool* pool()
	{
		FastMutex::ScopedLock lock(m_mutex);
		
		if (!m_pPool)
		{
			m_pPool = new ThreadPool("default");
			if (POCO_THREAD_STACK_SIZE > 0)
				m_pPool->setStackSize(POCO_THREAD_STACK_SIZE);
		}
		return m_pPool;
	}
	
private:
	ThreadPool* m_pPool;
	FastMutex   m_mutex;
};


namespace
{
	static ThreadPoolSingletonHolder sh;
}


ThreadPool& ThreadPool::defaultPool()
{
	return *sh.pool();
}


} // namespace Poco
